package com.lgd.experiment;



import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

import java.text.SimpleDateFormat;
import java.util.Date;




public class HbaseBatchImport {

    static Configuration conf = null;

    static {

        conf = HBaseConfiguration.create();


    }


    static class BatchImportMapper extends Mapper<LongWritable, Text, LongWritable, Text>{

        SimpleDateFormat dateformat1=new SimpleDateFormat("yyyyMMddHHmmss");

        Text v2 = new Text();


        protected void map(LongWritable key, Text value, Context context) throws java.io.IOException ,InterruptedException {

            final String[] splited = value.toString().split(",");

            try {

                final Date date = new Date(Long.parseLong(splited[0].trim()));

                final String dateFormat = dateformat1.format(date);

                String rowKey = splited[1]+":"+dateFormat;

                v2.set(rowKey+","+value.toString());

                context.write(key, v2);

            } catch (NumberFormatException e) {

                final Counter counter = context.getCounter("BatchImport", "ErrorFormat");

                counter.increment(1L);

                System.out.println("error"+splited[0]+" "+e.getMessage());

            }

        };

    }


    static class BatchImportReducer extends TableReducer<LongWritable, Text, NullWritable>{

        protected void reduce(LongWritable key, java.lang.Iterable<Text> values, Context context) throws java.io.IOException ,InterruptedException {

            for (Text text : values) {

                final String[] splited = text.toString().split(",");

                for(int i=0;i<splited.length;i++){

                    System.out.println(splited[i]);

                }

                final Put put = new Put(Bytes.toBytes(splited[0]));

                put.add(Bytes.toBytes("cf"), Bytes.toBytes("col1"), Bytes.toBytes(splited[1]));

                put.add(Bytes.toBytes("cf"), Bytes.toBytes("col2"), Bytes.toBytes(splited[2]));

                put.add(Bytes.toBytes("cf"),Bytes.toBytes("col3"), Bytes.toBytes(splited[3]));

//                put.add(Bytes.toBytes("f1"),Bytes.toBytes("col4"), Bytes.toBytes(splited[4]));

                context.write(NullWritable.get(), put);

            }

        };

    }


    public static void main(String[] args) throws Exception {




        conf.set("hbase.zookeeper.quorum", "ubuntu:2128");



        conf.set(TableOutputFormat.OUTPUT_TABLE, "test");




        conf.set("dfs.socket.timeout", "180000");


        final Job job = new Job(conf, "HBaseBatchImport");


        job.setMapperClass(BatchImportMapper.class);

        job.setReducerClass(BatchImportReducer.class);




        job.setMapOutputKeyClass(LongWritable.class);

        job.setMapOutputValueClass(Text.class);


        job.setInputFormatClass(TextInputFormat.class);




        job.setOutputFormatClass(TableOutputFormat.class);


        FileInputFormat.setInputPaths(job, "hdfs://ubuntu:9000/hbase/test1.txt");


        job.waitForCompletion(true);

    }

}